package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/3/1
 *
 *  广播状态，用于更新配置信息。
 *
 *      广播状态的含义： 指状态中存储的信息，一旦发生了变更，会自动广播到使用这个状态的所有的Task！
 */
public class Demo3_BroadcastState
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //主流
        SingleOutputStreamOperator<WaterSensor> ds1 = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //配置流 存放配置信息
        SingleOutputStreamOperator<MyConf> ds2 = env
            .socketTextStream("hadoop103", 8889)
            .map(x -> {
                String[] words = x.split(",");
                return new MyConf(words[0], words[1]);
            });

        //默认情况，ds2是一个普通的流，无法进行广播。必须把它变为广播流
        MapStateDescriptor<String, MyConf> mapStateDescriptor = new MapStateDescriptor<>("conf", String.class, MyConf.class);
        BroadcastStream<MyConf> broadcastStream = ds2.broadcast(mapStateDescriptor);

        //将主流和广播流连接
        ds1.connect(broadcastStream)
           .process(new BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>()
           {
               // BroadcastProcessFunction 的 processElement可以接受广播状态的方法
               @Override
               public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<WaterSensor> out) throws Exception {
                        //从广播状态中取
                   MyConf myConf = ctx.getBroadcastState(mapStateDescriptor).get(value.getId());
                        //更新配置
                        value.setId(myConf.getName());
                        out.collect(value);

               }

               @Override
               public void processBroadcastElement(MyConf value, Context ctx, Collector<WaterSensor> out) throws Exception {
                   //一旦有新的配置信息产生，就需要放入广播状态。 不用管了，自动广播到processElement方法中！
                   BroadcastState<String, MyConf> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                   broadcastState.put(value.getId(),value);

               }
           })
           .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyConf {
        private String id;
        private String name;
    }
}
